package com.stone.customer.console.trigger;

import com.stone.message.adapter.domain.bo.TriggerMessageParam;
import com.stone.message.adapter.infra.publisher.MessagePublisher;
import com.stone.message.adapter.trigger.AbstractCommonReceiverTrigger;
import com.stone.starter.redis.client.RedisCacheClient;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

/**
 * redis数据接收公用触发器
 * 对AbstractCommonReceiverTrigger进行默认实现，RedisReceiverCommonTrigger触发器适用于每个
 * 要从Redis取数据的触发器，使用的时候只需要注入RedisReceiverCommonTrigger类并且构造三个参数
 * （mqQueue、redisQueue、threadCount）然后传入executeTask（）方法即可实现从Redis指定的队列拉取数据。
 * 默认的失败补偿已经在compensate中实现好了，且默认是多线程异步执行任务（需要同步执行只需要更改isAsyncExecute方法范返回即可）
 * 具体使用见：RedisSyncTriggerTestImpl
 *
 * @author Mr_wenpan@163.com 2021/07/15 14:55
 */
@Slf4j
@Component
public class RedisReceiverCommonTrigger extends AbstractCommonReceiverTrigger<String, String> {

    /**
     * redis服务客户端
     */
    @Autowired
    private RedisCacheClient redisCacheClient;

    @Autowired
    @Qualifier("receiverFromRedisTriggerPool")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    public RedisReceiverCommonTrigger(MessagePublisher messagePublisher) {
        super(messagePublisher);
    }

    /**
     * 执行触发任务
     *
     * @param mqQueue     取出来的数据放入到哪个mq
     * @param redisQueue  从Redis哪个queue取数据
     * @param threadCount 并发线程数
     * @author Mr_wenpan@163.com 2021/7/17 2:16 下午
     */
    public void executeTask(String mqQueue, String redisQueue, Integer threadCount) {
        TriggerMessageParam<String> param = TriggerMessageParam.<String>builder()
                .queue(mqQueue)
                .queryDataParam(redisQueue)
                .taskCount(threadCount)
                .build();
        triggerWithCheck(param, this::compensate);
    }

    /**
     * 触发器触发执行参数由用户自己构建
     *
     * @param triggerMessageParam 触发参数
     * @author Mr_wenpan@163.com 2021/7/17 2:36 下午
     */
    public void executeTask(TriggerMessageParam<String> triggerMessageParam) {
        triggerWithCheck(triggerMessageParam, this::compensate);
    }

    @Override
    protected ThreadPoolTaskExecutor threadPoolTaskExecutor() {
        return threadPoolTaskExecutor;
    }

    @Override
    protected String getData(TriggerMessageParam<String> param) {
        log.info("获取redis数据入参：{}", param);
        return redisCacheClient.opsForList().rightPop(param.getQueryDataParam());
    }

    /**
     * 开启多线程执行
     */
    @Override
    protected boolean isAsyncExecute() {
        return Boolean.TRUE;
    }

    /**
     * 通用错误补偿方法
     *
     * @param data  需要补偿的数据
     * @param param 参数
     * @author Mr_wenpan@163.com 2021/7/17 2:07 下午
     */
    private void compensate(String data, TriggerMessageParam<String> param) {
        // 写入数据库的错误记录表，记录出错数据、出错数据所对应的Redis队列的key，出错时间，出错类型，错误次数等
        // 出错类型: 有可能是多种错误情况写入同一张表，比如从Redis消费的错误，或job执行错误等
        // 错误次数: 方便多次重试还是出错的情况发送告警邮件，人工干预检查是否数据有问题
        String redisQueue = param.getQueryDataParam();
        String mqQueue = param.getQueue();

        log.error("从Redis获取数据同步到数据库出错啦！redisQueue={},mqQueue={},data={}", redisQueue, mqQueue, data);

        // todo 错误信息写入数据库错误记录表

    }
}
